package com.apps.sdses.flink141.sql.window;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/7/5 10:26
 */
public class TumbleWindowsDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Configuration configuration = tableEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "GMT-08:00");

        //Source
        tableEnv.executeSql("CREATE TABLE sourceTable (\n" +
                "    id        BIGINT,\n" +     //采集信息id
                "    sid        STRING,\n" +    //水位传感器id
                "    ts   TIMESTAMP(3),\n" +    //数据采集时间
                "    flag   BOOLEAN,\n" +       //水位报警开关
                "    vc INT,\n" +                //采集到的水位高度
                "    WATERMARK FOR ts AS ts - INTERVAL '10' SECOND\n" +                //采集到的水位高度
                ") WITH (\n" +
                "  'connector' = 'datagen', \n" +
//                "  'number-of-rows' = '100000', \n" +   //每秒生成1条数据
                "  'rows-per-second' = '10', \n" +   //每秒生成1条数据

                "  'fields.sid.length' = '5', \n" +   //传感器id长度

//                "  'fields.id.kind' = 'sequence', \n" +
                "  'fields.id.min' = '1', \n" +
                "  'fields.id.max' = '100', \n" +

                "  'fields.vc.min' = '10', \n" +
                "  'fields.vc.max' = '100'\n" +
                ")");

//        tableEnv.executeSql("set 'table.exec.sink.not-null-enforcer'='drop'\n");

//        tableEnv.executeSql("CREATE TABLE sinkTable2 WITH ('connector' = 'print')\n" +
//                "LIKE sourceTable (EXCLUDING ALL)");
//        tableEnv.executeSql("insert into sinkTable2 select * from sourceTable");


        tableEnv.executeSql("CREATE TABLE sinkTable (\n" +
                " window_start TIMESTAMP, \n" +
                " window_end TIMESTAMP, \n" +
                " window_time TIMESTAMP, \n" +
                " agg_number BIGINT) \n" +
                " with ('connector' = 'print')");

        //Insert
        //滚动事件窗口
        tableEnv.executeSql("insert into sinkTable select \n" +
                "window_start, window_end, window_time, sum(vc) \n" +
                "from TABLE(TUMBLE(TABLE sourceTable, DESCRIPTOR(ts), INTERVAL '5' SECONDS)) \n" +
                "GROUP BY window_start, window_end, window_time");


//        //滑动时间窗口
//        tableEnv.executeSql("insert into sinkTable select \n" +
//                "window_start" +
//                "" +
//                "")





    }
}
